package flink_p1

import org.apache.flink.api.common.serialization.SimpleStringSchema
import org.apache.flink.api.common.state.{BroadcastState, MapStateDescriptor, ReadOnlyBroadcastState}
import org.apache.flink.api.common.typeinfo.BasicTypeInfo
import org.apache.flink.streaming.api.datastream.{BroadcastConnectedStream, BroadcastStream, DataStream}
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment
import org.apache.flink.streaming.api.functions.co.BroadcastProcessFunction
import org.apache.flink.streaming.connectors.kafka.FlinkKafkaConsumer
import org.apache.flink.util.Collector
import org.apache.kafka.clients.consumer.ConsumerConfig
import org.apache.kafka.common.serialization.StringDeserializer

import java.util.Properties

object FlinkTest_22_table3_bc_kafkaConfig {

  /**
   * 关联维度表：通过kafka更新维度表
   * 将kafka的维度信息通过广播分发给下游分区，得到广播流
   * 再将广播流与数据流进行connect
   *
   * @param args
   */
  def main(args: Array[String]): Unit = {

    val env: StreamExecutionEnvironment = StreamExecutionEnvironment.getExecutionEnvironment


    val socketStream: DataStream[String] = env.socketTextStream("127.0.0.1", 8889)


    val prop = new Properties()
    prop.setProperty(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG, "127.0.0.1:9092")
    prop.setProperty(ConsumerConfig.KEY_DESERIALIZER_CLASS_CONFIG, classOf[StringDeserializer].getName)
    prop.setProperty(ConsumerConfig.VALUE_DESERIALIZER_CLASS_CONFIG, classOf[StringDeserializer].getName)
    prop.setProperty(ConsumerConfig.GROUP_ID_CONFIG, "g1")


    val kafkaConsumer: FlinkKafkaConsumer[String] = new FlinkKafkaConsumer[String]("dy_config_topic", new SimpleStringSchema(), prop)

    kafkaConsumer.setStartFromLatest()

    val configSource: DataStream[String] = env.addSource(kafkaConsumer)

    val dyConfigMapDesc: MapStateDescriptor[String, String] = new MapStateDescriptor[String, String]("dyConfigMap", BasicTypeInfo.STRING_TYPE_INFO, BasicTypeInfo.STRING_TYPE_INFO)

    val bcStream: BroadcastStream[String] = configSource.broadcast(dyConfigMapDesc)

    val bcConnStream: BroadcastConnectedStream[String, String] = socketStream.connect(bcStream)

    bcConnStream.process(new BroadcastProcessFunction[String, String, (String, String)] {
      override def processElement(value: String, ctx: BroadcastProcessFunction[String, String, (String, String)]#ReadOnlyContext, out: Collector[(String, String)]): Unit = {

        val bcMapState: ReadOnlyBroadcastState[String, String] = ctx.getBroadcastState(dyConfigMapDesc)
        val text: String = bcMapState.get(value)

        if (text != null) {
          out.collect((value, text))
        } else {
          out.collect((value, "not text"))
        }


      }

      override def processBroadcastElement(value: String, ctx: BroadcastProcessFunction[String, String, (String, String)]#Context, out: Collector[(String, String)]): Unit = {

        val bcState: BroadcastState[String, String] = ctx.getBroadcastState(dyConfigMapDesc)

        val arr: Array[String] = value.split(" ")
        bcState.put(arr(0), arr(1))

      }
    }).print()


    env.execute()
  }

}
